package org.pliers.canal.config;

import org.pliers.canal.annotation.CanalSubscriber;
import org.pliers.canal.client.CanalClientFactory;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.core.Ordered;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.util.ReflectionUtils;

import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

public class CanalSubscriberAnnotationBeanPostProcessor implements BeanPostProcessor, Ordered, SmartInitializingSingleton {

    private final ConcurrentMap<Class<?>, TypeMetadata<CanalSubscriber>> canalSubscribeCache = new ConcurrentHashMap<>();
    private final CanalClientFactory canalClientFactory  ;

    public CanalSubscriberAnnotationBeanPostProcessor(CanalClientFactory canalClientFactory) {
        this.canalClientFactory = canalClientFactory;
    }

    @Override
    public void afterSingletonsInstantiated() {
        canalClientFactory.start();
        //subscribe
        canalSubscribeCache.forEach((key, value) -> {
            Object target = value.target;
            for (ListenerMethod<CanalSubscriber> m : value.listenerMethods) {
                CanalSubscriber an = m.annotation;
                Method method = m.method;
                canalClientFactory.subscribe(an.schemaName(), an.tableName(), an.eventType() , (rowChange) -> ReflectionUtils.invokeMethod(method, target, rowChange));
            }
        });
        canalSubscribeCache.clear();
    }

    @Override
    public int getOrder() {
        return LOWEST_PRECEDENCE;
    }

    @Override
    public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
        return bean;
    }

    @Override
    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
        Class<?> targetClass = AopUtils.getTargetClass(bean);
        //initialize subscriber
        TypeMetadata<CanalSubscriber> metadata = buildMetadata(bean, targetClass, CanalSubscriber.class);
        if (metadata != null) {
            canalSubscribeCache.putIfAbsent(targetClass, metadata);
        }
        return bean ;
    }

    private static class TypeMetadata<T> {
        final Object target;

        final ListenerMethod<T>[] listenerMethods;

        TypeMetadata(Object target, ListenerMethod<T>[] methods) {
            this.target = target;
            this.listenerMethods = methods;
        }
    }

    private static class ListenerMethod<T> {
        final Method method;

        final T annotation;

        ListenerMethod(Method method, T annotation) {
            this.method = method;
            this.annotation = annotation;
        }
    }

    private <T extends Annotation> TypeMetadata<T> buildMetadata(Object bean, Class<?> targetClass, Class<T> annotationClass) {
        final List<ListenerMethod<T>> methods = new ArrayList<>();
        ReflectionUtils.doWithMethods(targetClass, method -> {
            T ms = findListenerAnnotations(method, annotationClass);
            if (ms != null) {
                methods.add(new ListenerMethod<>(method, ms));
            }
        }, ReflectionUtils.USER_DECLARED_METHODS);

        if (methods.isEmpty()) {
            return null;
        }
        @SuppressWarnings("unchecked")
        ListenerMethod<T>[] array = (ListenerMethod<T>[]) new ListenerMethod[methods.size()];

        return new TypeMetadata<>(bean, methods.toArray(array));
    }

    private <T extends Annotation> T findListenerAnnotations(Method method, Class<T> targetClass) {
        return AnnotationUtils.findAnnotation(method, targetClass);
    }
}
